package com.beiding.mpTimer;

import java.io.Serializable;
import java.util.Map;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 时间轴对象实现
 */

public class TimerShaftImpl implements TimerShaft {

    //当作为时间轴持有者时,会经过指定的时间进行一次跳动,检查对应时间点上是否有任务,类似于钟表上的指针
    private Timer pointer;

    //派发任务执行线程池
    private ExecutorService distributeExecutor;

    /**
     * 心跳线程
     * <p>
     * 当作为时间轴持有者时:
     * 1.检查过期任务
     * 2.修改最后一次心跳时间
     * <p>
     * 当作为非时间轴持有者时:
     * 判断时间轴持有者是否依然存活,如果已经死亡则取而代之
     */
    private Timer heartbeat;

    //时间轴服务
    private TimerShaftService service;

    //时间轴对象id
    private String id;

    //指针步长和心跳间隔
    private long step, heartbeatInterval;

    //当前时间点.由指针线程给出
    private long current;

    //准时任务派发器及过期任务派发器
    private Dispatcher onTimeDispatcher, overdueTasksDispatcher;

    //使用默认的id生成器策略
    private IdCreateStrategy idCreateStrategy;

    //可以自定义id生成策略
    public void setIdCreateStrategy(IdCreateStrategy idCreateStrategy) {
        this.idCreateStrategy = idCreateStrategy;
    }

    public IdCreateStrategy getIdCreateStrategy() {
        return idCreateStrategy;
    }

    //过时任务和准时任务公用一个派发器
    public TimerShaftImpl(TimerShaftService timerShaftService, long step, long heartbeatInterval, Dispatcher dispatcher) {
        this(timerShaftService, step, heartbeatInterval, dispatcher, dispatcher);
    }

    //过时任务派发器和准时任务公用一个派发器并指定id生成策略
    public TimerShaftImpl(TimerShaftService timerShaftService, long step, long heartbeatInterval, Dispatcher dispatcher, IdCreateStrategy idCreateStrategy) {
        this(timerShaftService, step, heartbeatInterval, dispatcher, dispatcher, idCreateStrategy);
    }

    public TimerShaftImpl(TimerShaftService timerShaftService, long step, long heartbeatInterval, Dispatcher onTimeDispatcher, Dispatcher overdueTasksDispatcher) {
        this(timerShaftService, step, heartbeatInterval, onTimeDispatcher, overdueTasksDispatcher, new IdCreateStrategy() {
        });

    }

    public TimerShaftImpl(TimerShaftService timerShaftService, long step, long heartbeatInterval, Dispatcher onTimeDispatcher, Dispatcher overdueTasksDispatcher, IdCreateStrategy idCreateStrategy) {
        this.idCreateStrategy = idCreateStrategy;
        this.service = timerShaftService;
        this.step = step;
        this.heartbeatInterval = heartbeatInterval;
        this.onTimeDispatcher = onTimeDispatcher;
        this.overdueTasksDispatcher = overdueTasksDispatcher;

        init();

    }


    //初始化
    private void init() {

        if (step < 200) {
            throw new IllegalArgumentException("步长过短");
        }

        if (heartbeatInterval < 2 * step) {
            throw new IllegalArgumentException("心跳间隔至少是步长的两倍");
        }


        id = "timerShaft_" + hashCode() + "_" + System.currentTimeMillis();

        String holderId = service.getTimerShaftHolderId();

        //如果没有时间轴对象
        if (holderId == null) {

            long theStep = service.getStep();

            long theHeartbeatInterval = service.getHeartbeatInterval();

            //如果两个值都未指定,则指定
            if (theStep == 0 && theHeartbeatInterval == 0) {

                service.setStep(step);
                service.setHeartbeatInterval(heartbeatInterval);

            } else if (theStep != step || theHeartbeatInterval != heartbeatInterval) {

                throw new IllegalArgumentException("步长和心跳间隔只能指定一次.当前步长:" + theStep + ",当前心跳间隔:" + theHeartbeatInterval);
            }

            tryHold();
        } else {
            asDefender();
        }

    }

    //尝试持有时间轴
    private void tryHold() {

        //首先修改时间轴持有者为自己
        service.setTimerShaftHolderId(id);


        if (distributeExecutor == null) {
            distributeExecutor = Executors.newSingleThreadExecutor();
        }


        //更新心跳线程的行为
        if (heartbeat != null) {
            heartbeat.cancel();
        }

        heartbeat = new Timer();

        heartbeat.schedule(new TimerTask() {
            @Override
            public void run() {

                //更新最后一次心跳时间
                service.updateLastHeartbeatTimestamp(System.currentTimeMillis());

                //获取过期任务.获取小于当前时刻的任务
                Map<Long, Iterable<String>> tasks = service.getOverdueTasks(current);

                distributeExecutor.submit(() -> {

                    //过期任务派发
                    tasks.forEach((aLong, strings) -> {
                        overdueTasksDispatcher.distribute(aLong, strings);
                    });

                });

                String timerShaftHolderId = service.getTimerShaftHolderId();

                //如果丢失时间轴
                if (!id.equals(timerShaftHolderId)) {

                    //成为守护者
                    asDefender();
                }

            }
        }, 0, heartbeatInterval);


        //更新指针线程行为
        if (pointer != null) {
            pointer.cancel();
        }

        pointer = new Timer();

        pointer.schedule(new TimerTask() {
            @Override
            public void run() {

                current = System.currentTimeMillis() / step;

                //获取准时任务
                Iterable<String> tasks = service.getOnTimeTasks(current);

                //进行任务派发
                onTimeDispatcher.distribute(current, tasks);

            }
        }, 0, step);

    }

    //作为持有者的守护者
    private void asDefender() {

        //更新心跳线程的行为
        if (heartbeat != null) {
            heartbeat.cancel();
        }

        heartbeat = new Timer();

        heartbeat.schedule(new TimerTask() {
            @Override
            public void run() {

                //获取最后一次心跳时间戳
                long lastHeartbeatTime = service.getLastHeartbeatTimestamp();

                long currentTimeMillis = System.currentTimeMillis();

                //基本可以断言记录的最后一次心跳时间与当前事件不会相差一个心跳间隔
                if (currentTimeMillis - lastHeartbeatTime > heartbeatInterval) {
                    tryHold();
                }

            }
        }, 0, heartbeatInterval);


        //更新指针线程行为
        if (pointer != null) {
            pointer.cancel();
        }

        pointer = null;

        //停止任务派发器
        if (distributeExecutor != null) {
            distributeExecutor.shutdown();
            distributeExecutor = null;
        }

    }

    @Override
    public String delay(Serializable payload, String topic, long duration) {

        long tm = System.currentTimeMillis() + duration;

        //发生时间点由发生时间的毫秒数除以步长加一得到
        long time = tm / step + 1;

        TaskPersistence persistence = new TaskPersistence(idCreateStrategy.createId(payload, topic), payload, topic, time, tm);

        service.permanent(persistence);

        return persistence.getId();

    }

    @Override
    public String onTime(Serializable payload, String topic, long time) {

        //发生时间点有发生时间毫秒数除以步长加一得到
        long t = time / step + 1;

        TaskPersistence persistence = new TaskPersistence(idCreateStrategy.createId(payload, topic), payload, topic, t, time);

        service.permanent(persistence);

        return persistence.getId();

    }

    @Override
    public Object invalidate(String tackId) {
        return service.remove(tackId);
    }

    //使用id生成策略生成对应的id再移除任务
    @Override
    public void invalidate(Serializable payload, String topic) {
        service.remove(idCreateStrategy.createId(payload, topic));
    }
}
